package avicit.bdp.dcs.json.service;

import avicit.bdp.common.dto.api.APIHttpDefinition;
import avicit.bdp.common.utils.DateUtils;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.constant.ProcessConstants;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.task.service.ApiDefinitionService;
import avicit.bdp.dcs.task.service.TaskService;
import avicit.bdp.dcs.tools.ds.NodeConnect;
import avicit.bdp.dcs.tools.enums.Enable;
import avicit.bdp.dcs.utils.DcsConstants;
import avicit.bdp.dds.dispatch.enums.*;
import avicit.bdp.dds.dispatch.model.ProcessData;
import avicit.bdp.dds.dispatch.model.TaskNode;
import avicit.bdp.dds.dispatch.process.Property;
import avicit.bdp.dds.dispatch.task.conditions.ConditionsParameters;
import avicit.bdp.dds.dispatch.task.datax.DataxParameters;
import avicit.bdp.dds.dispatch.task.http.HttpParameters;
import com.alibaba.fastjson2.JSON;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.*;

/**
 * @author LIYB json构建实现类
 */
@Service
@Transactional
public class ApiTaskJsonService extends TaskJsonService{

  private static final Logger logger = LoggerFactory.getLogger(ApiTaskJsonService.class);

  @Resource private TaskService taskService;
  @Resource private ApiDefinitionService apiDefinitionService;

  /**
   * 构建API采集流程定义数据
   *
   * @param jobDTO
   * @return
   * @throws Exception
   */
  public Map<String, String> buildDsJson(JobDTO jobDTO) throws Exception {
    Map<String, String> map = new HashMap<>(4);

    // 查询job下的任务列表
    TaskDTO taskDTOQuery = new TaskDTO();
    taskDTOQuery.setJobId(jobDTO.getId());
    taskDTOQuery.setFlag(Enable.ZERO.getCode());
    List<TaskDTO> taskList = taskService.selectList(taskDTOQuery);

    if (CollectionUtils.isNotEmpty(taskList)) {
      ProcessData processData = new ProcessData();
      List<TaskNode> taskNodeList = new LinkedList<>();
      List<NodeConnect> nodeConnectList = new LinkedList<>();
      String endTime = DateUtils.getDateByString(new Date());

      List<List<TaskDTO>> blocks = splitIntoBlocks(taskList);

      for(List<TaskDTO> list : blocks){
        TaskNode taskNode = null;
        for (TaskDTO task : list) {
          taskNode = createDataxNodeByApi(jobDTO, task, taskNodeList, endTime, nodeConnectList, taskNode);
        }
      }

      List<Property> extraParams = new ArrayList<>();
      Property dateIntervalProperty = new Property();
      dateIntervalProperty.setProp(DcsConstants.DATA_END_TIME);
      dateIntervalProperty.setDirect(Direct.IN);
      dateIntervalProperty.setType(DataType.VARCHAR);
      dateIntervalProperty.setValue(endTime);
      extraParams.add(dateIntervalProperty);

      processData.setTasks(taskNodeList);
      map.put(ProcessConstants.PROCESS_DEFINITION_JSON, JSON.toJSONString(processData));
      map.put(ProcessConstants.CONNECTS, JSON.toJSONString(nodeConnectList));
      map.put(ProcessConstants.EXTRA_PARAMS, JSON.toJSONString(extraParams));
      logger.info(
              "************processDefinitionJson长度={}", JSON.toJSONString(processData).length());
    }
    return map;
  }

  /** 创建Datax节点 */
  private TaskNode createDataxNodeByApi(
          JobDTO jobDTO,
          TaskDTO task,
          List<TaskNode> taskNodeList,
          String endTime,
          List<NodeConnect> nodeConnectList,
          TaskNode lastTaskNode)
          throws Exception {
    Random random = new Random();
    List<String> preTaskList = new ArrayList<>();

    // 创建前置处理节点
    String preProcessNodeId =
            ProcessConstants.TASKS + Constants.SUBTRACT_STRING + random.nextInt(MAX_RANDOM);
    TaskNode preProcessNode = new TaskNode();
    preProcessNode.setId(preProcessNodeId);
    preProcessNode.setType(TaskType.HTTP.name());
    preProcessNode.setName(task.getId() + Constants.SUBTRACT_STRING + PRE_PROCESS_NODE);
    preProcessNode.setDesc(Constants.STRING_BLANK);
    preProcessNode.setRunFlag(ProcessConstants.NORMAL);
    preProcessNode.setMaxRetryTimes(0);
    preProcessNode.setRetryInterval(1);
    preProcessNode.setTaskInstancePriority(Priority.MEDIUM);
    preProcessNode.setWorkerGroup(ProcessConstants.DEFAULT);
    preProcessNode.setConditionResult(JSON.toJSONString(new ConditionsParameters()));

    HttpParameters httpParameters = new HttpParameters();
    httpParameters.setUrl(preProcessUrl + Constants.SINGLE_SLASH + task.getId());
    httpParameters.setHttpMethod(HttpMethod.GET);
    preProcessNode.setParams(JSON.toJSONString(httpParameters));
    taskNodeList.add(preProcessNode);

    // 任务串行执行
    if(lastTaskNode != null){
      NodeConnect preProcessNodeConnect = new NodeConnect();
      preProcessNodeConnect.setEndPointSourceId(lastTaskNode.getId());
      preProcessNodeConnect.setEndPointTargetId(preProcessNodeId);
      nodeConnectList.add(preProcessNodeConnect);

      preTaskList = new ArrayList<>();
      preTaskList.add(lastTaskNode.getName());
      preProcessNode.setPreTasks(JSON.toJSONString(preTaskList));
    }

    // 创建dataX采集数据节点
    String taskNodeId =
            ProcessConstants.TASKS + Constants.SUBTRACT_STRING + random.nextInt(MAX_RANDOM);
    TaskNode taskNode = new TaskNode();
    taskNode.setId(taskNodeId);

    buildDataxNodeByApi(taskNode, jobDTO, task, endTime);

    // 需要创建表结构，则创建HTTP类型的前置节点，该节点用于回调数据采集系统中创建表服务
    boolean createTable = true;
    if (createTable) {
      String createTableNodeId =
              ProcessConstants.TASKS + Constants.SUBTRACT_STRING + random.nextInt(MAX_RANDOM);
      TaskNode createTableNode = new TaskNode();
      createTableNode.setId(createTableNodeId);
      createTableNode.setType(TaskType.HTTP.name());
      createTableNode.setName(task.getId() + Constants.SUBTRACT_STRING + CREATE_TABLE_NODE);
      createTableNode.setDesc(Constants.STRING_BLANK);
      createTableNode.setRunFlag(ProcessConstants.NORMAL);
      createTableNode.setMaxRetryTimes(0);
      createTableNode.setRetryInterval(1);
      createTableNode.setTaskInstancePriority(Priority.MEDIUM);
      createTableNode.setWorkerGroup(ProcessConstants.DEFAULT);

      createTableNode.setConditionResult(JSON.toJSONString(new ConditionsParameters()));

      httpParameters = new HttpParameters();
      httpParameters.setUrl(createTableUrl + Constants.SINGLE_SLASH + task.getId());
      httpParameters.setHttpMethod(HttpMethod.GET);
      createTableNode.setParams(JSON.toJSONString(httpParameters));
      taskNodeList.add(createTableNode);

      // 构建Dolphinscheduler流程定义节点连接信息json,[{"endPointSourceId":"tasks-44147","endPointTargetId":"tasks-70818"}]
      NodeConnect nodeConnect = new NodeConnect();
      nodeConnect.setEndPointSourceId(preProcessNodeId);
      nodeConnect.setEndPointTargetId(createTableNodeId);
      nodeConnectList.add(nodeConnect);

      preTaskList = new ArrayList<>();
      preTaskList.add(preProcessNode.getName());
      createTableNode.setPreTasks(JSON.toJSONString(preTaskList));

      nodeConnect = new NodeConnect();
      nodeConnect.setEndPointSourceId(createTableNodeId);
      nodeConnect.setEndPointTargetId(taskNodeId);
      nodeConnectList.add(nodeConnect);

      preTaskList = new ArrayList<>();
      preTaskList.add(createTableNode.getName());
      taskNode.setPreTasks(JSON.toJSONString(preTaskList));
    } else {
      NodeConnect nodeConnect = new NodeConnect();
      nodeConnect.setEndPointSourceId(preProcessNodeId);
      nodeConnect.setEndPointTargetId(taskNodeId);
      nodeConnectList.add(nodeConnect);

      preTaskList = new ArrayList<>();
      preTaskList.add(preProcessNode.getName());
      taskNode.setPreTasks(JSON.toJSONString(preTaskList));
    }
    taskNodeList.add(taskNode);
    return taskNode;
  }

  /**
   * 构造DATAX采集节点
   *
   * @param taskNode
   * @param jobDTO
   * @param task
   * @param endTime
   */
  private void buildDataxNodeByApi(TaskNode taskNode, JobDTO jobDTO, TaskDTO task, String endTime) {
    taskNode.setType(TaskType.DATAX.name());
    taskNode.setName(task.getId());
    taskNode.setDesc(Constants.STRING_BLANK);
    taskNode.setRunFlag(ProcessConstants.NORMAL);
    taskNode.setMaxRetryTimes(0);
    taskNode.setRetryInterval(1);
    taskNode.setTaskInstancePriority(Priority.MEDIUM);
    taskNode.setWorkerGroup(ProcessConstants.DEFAULT);

    ConditionsParameters conditionsParameters = new ConditionsParameters();
    taskNode.setConditionResult(JSON.toJSONString(conditionsParameters));

    DataxParameters dataxParameters = new DataxParameters();
    dataxParameters.setCustomConfig(1);
    dataxParameters.setJson(task.getDataxJson());
    dataxParameters.setDataSource(jobDTO.getSrcDatasourceId());
    dataxParameters.setSourceTable(task.getSrcTableName());
    dataxParameters.setDataTarget(jobDTO.getDstDatasourceId());
    dataxParameters.setTargetTable(task.getDstTableName());
    dataxParameters.setColumnSource(getSourceColumn(task));
    dataxParameters.setColumnTarget(getDstColumn(task));

    // 如果是增量同步方式，则添加自定义属性
    APIHttpDefinition apiHttpDefinition = apiDefinitionService.getAPIHttpDefinition(jobDTO.getId());
    List<Property> localParams = new ArrayList<>();

    if (apiHttpDefinition.isHaveIncrement()) {
      String startTime = apiHttpDefinition.getIncrementParam().getInitialValue();
      // 增量同步的开始时间戳
      Property startProperty = new Property();
      startProperty.setProp(ProcessConstants.DATA_START_TIME);
      startProperty.setDirect(Direct.IN);
      startProperty.setType(DataType.TIMESTAMP);
      startProperty.setValue(startTime);
      localParams.add(startProperty);

      // 增量同步的结束时间戳
      Property endProperty = new Property();
      endProperty.setProp(ProcessConstants.DATA_END_TIME);
      endProperty.setDirect(Direct.IN);
      endProperty.setType(DataType.TIMESTAMP);
      endProperty.setValue(endTime);
      localParams.add(endProperty);
    }

    int jobType = 2;
    if (apiHttpDefinition.isHaveIncrement()) {
      jobType = 3;
    } else {
      jobType = 2;
    }

    Property property = new Property();
    property.setProp(DcsConstants.JOB_TYPE);
    property.setDirect(Direct.IN);
    property.setType(DataType.VARCHAR);
    property.setValue(String.valueOf(jobType));
    localParams.add(property);

    dataxParameters.setLocalParams(localParams);

    taskNode.setParams(JSON.toJSONString(dataxParameters));
  }

}
